#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：解析离线数据.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/5/26 8:42
'''
import json
# 1:26    2:25
import time
from multiprocessing import Process

from kafka import KafkaProducer

from utils.tools import time_str, time_str_jingtai

KAFKA_HOST = '10.100.12.58:9093'
def send_kafka(file_path, topic):
    file_path = f'C:/Users/xinyingjie/Downloads/kafka-logs' +'/'+ file_path + "/" + "00000000000000000000.log"
    kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                   value_serializer=lambda x: json.dumps(x).encode('utf-8'),
                                   compression_type='gzip')
    before_time = None
    with open(file_path, 'r', encoding='latin-1') as f:
        while True:
            line = f.readline().strip()
            if line:
                print(line[:300])
                if 'e1_data:' in line:

                    temp = line.split("e1_data:")[1]

                    data = temp[temp.find("{"):temp.rfind("]}") + 2].replace("'", '"')
                    # print(data)
                    json_data = None
                    try:
                        json_data = json.loads(data)
                    except Exception as e:
                        if data:
                            json_data = json.loads(data[1:])
                    if json_data:
                        json_data['globalTime'] = time_str_jingtai(int(time.time() * 1000))
                        print(json_data['globalTime'])
                        print(json_data)
                        kafka_producer.send(topic, value=json_data,
                                            key=str(int(time.time())))

                        time.sleep(1)
            else:
                continue

#
# file_path_list = ['e1_data_CJ-10-01-0', 'e1_data_CJ-10-02-0', 'e1_data_JJ-10-01-0', 'e1_data_JJ-10-02-0']
if __name__ == '__main__':

    file_path_list = ['e1_data_CJ-10-01-0', 'e1_data_CJ-10-02-0', 'e1_data_JJ-10-01-0', 'e1_data_JJ-10-02-0']
    for file_path in file_path_list:
        # send_kafka(file_path, file_path[:-2])
        tess_process = Process(target=send_kafka,
                               args=(file_path, file_path[:-2]))
        tess_process.start()
